package com.meiya.flink.processor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.meiya.flink.config.AppConfig;
import com.meiya.flink.interfaces.TagProcessor;

/**
 * 大额交易标签处理器
 * 识别单笔交易金额超过5千元的交易
 */
public class LargeAmountTradeTagProcessor implements TagProcessor {
    private static final Logger logger = LoggerFactory.getLogger(LargeAmountTradeTagProcessor.class);
    
    // 标签ID
    private static final Long MID = 1001L;
    
    // 标签名称
    private static final String TAG_NAME = "大额交易测试用";
    
    // 大额交易阈值（单位：元）
    private static final double LARGE_AMOUNT_THRESHOLD = 5000.0;

    @Override
    public Long getMId() {
        return MID;
    }

    @Override
    public String getTagName() {
        return TAG_NAME;
    }
    
    @Override
    public int getSessionWindowIntervalMinutes() {
        // 使用配置的会话窗口间隔时间
        return AppConfig.getSessionWindowIntervalMinutes();
    }

    @Override
    public String generateTagSQL() {
        int sessionWindowInterval = getSessionWindowIntervalMinutes();
        logger.info("大额交易标签使用会话窗口间隔: {} 分钟", sessionWindowInterval);
        
        // 生成标签计算SQL
        return "INSERT INTO tags_result\n" +
                "WITH processed_trades_data AS (\n" +
                "  SELECT clue_id, jykh, mid, last_process_time\n" +
                "  FROM processed_trades\n" +
                "  WHERE mid = '" + MID + "'\n" +
                "),\n" +
                "processed_tags_data AS (\n" +
                "  SELECT clue_id, jykh, mid, last_process_time\n" +
                "  FROM processed_tags\n" +
                "  WHERE mid = '" + MID + "'\n" +
                "),\n" +
                "existing_tags_data AS (\n" +
                "  SELECT mid, clue_id, jykh, tags, ukeys\n" +
                "  FROM doris_tags_result\n" +
                "  WHERE mid = " + MID + "\n" +
                "  AND tags = '" + TAG_NAME + "'\n" +
                "),\n" +
                "new_transactions AS (\n" +
                "  SELECT t.*\n" +
                "  FROM transaction_session_window_view t\n" +
                "  LEFT JOIN processed_trades_data pt ON t.clue_id = pt.clue_id AND t.jykh = pt.jykh\n" +
                "  LEFT JOIN processed_tags_data ptg ON t.clue_id = ptg.clue_id AND t.jykh = ptg.jykh\n" +
                "  WHERE t.jyje >= " + LARGE_AMOUNT_THRESHOLD + "\n" +
                "    AND (\n" +
                "      (pt.clue_id IS NULL) -- 没有处理过的交易\n" +
                "      OR\n" +
                "      (pt.clue_id IS NOT NULL AND ptg.clue_id IS NULL) -- 处理过交易但没有该标签\n" +
                "      OR\n" +
                "      (pt.clue_id IS NOT NULL AND ptg.clue_id IS NOT NULL AND NOT EXISTS (\n" +
                "        SELECT 1 FROM existing_tags_data etd\n" +
                "        WHERE etd.clue_id = t.clue_id AND etd.jykh = t.jykh\n" +
                "        AND (POSITION(t.ukey IN etd.ukeys) > 0 OR POSITION(CONCAT(',', t.ukey) IN etd.ukeys) > 0 OR POSITION(CONCAT(t.ukey, ',') IN etd.ukeys) > 0)\n" +
                "      )) -- 处理过交易且有该标签，但当前ukey未处理过\n" +
                "    )\n" +
                ")\n" +
                "SELECT\n" +
                "  " + MID + " AS mid,\n" +
                "  t.clue_id,\n" +
                "  '" + TAG_NAME + "' AS tags,\n" +
                "  t.jykh,\n" +
                "  t.jyfhm,\n" +
                "  t.jyfzjhm AS khzjhm,\n" +
                "  t.bank_name AS khyhmc,\n" +
                "  t.jyje AS zjyje,\n" +
                "  1 AS jycz,\n" +
                "  CURRENT_TIMESTAMP AS cjsj,\n" +
                "  CURRENT_TIMESTAMP AS xgsj,\n" +
                "  t.ukey AS ukeys\n" +
                "FROM new_transactions t\n" +
                "GROUP BY\n" +
                "  t.clue_id,\n" +
                "  t.jykh,\n" +
                "  t.jyfhm,\n" +
                "  t.jyfzjhm,\n" +
                "  t.bank_name,\n" +
                "  t.jyje,\n" +
                "  t.ukey,\n" +
                "  t.window_start,\n" +
                "  t.window_end";
    }
} 